package iotdb

import (
	"errors"
	"fmt"
	"log"
	"math/rand"
	"reflect"
	"strconv"
	"sync"
	"time"

	"github.com/apache/iotdb-client-go/client"
	"github.com/apache/iotdb-client-go/rpc"
	"github.com/nick/vdamtools/apps/config"
	"github.com/nick/vdamtools/apps/model"
	"github.com/nick/vdamtools/apps/monitor"
	"github.com/nick/vdamtools/pkg/loggerv"
	"go.uber.org/zap"
)

// https://raw.githubusercontent.com/apache/iotdb-client-go/main/example/session_pool/session_pool_example.go

type IoTDB struct {
	sessionPool *client.SessionPool
}

var once sync.Once //确保只会执行一次
var instance *IoTDB

func GetInstance() *IoTDB {
	once.Do(func() {
		loggerv.GetLogger().Info("get config:", zap.Any("GlobalConfig", config.GlobalConfig))
		db := config.GlobalConfig.IotDb
		loggerv.GetLogger().Info("get config:", zap.Any("iot", db))
		config := client.PoolConfig{
			Host:     db.Host,
			Port:     db.Port,
			UserName: db.UserName,
			Password: db.Password,
		}
		pool := client.NewSessionPool(&config, 3, 60000, 60000, false)
		instance = &IoTDB{sessionPool: &pool}
	})
	return instance
}

func (db *IoTDB) Close() {
	db.sessionPool.Close()
}

func (db *IoTDB) SetStorageGroup(sg string) {
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		session.SetStorageGroup(sg)
	}
}

func (db *IoTDB) CountTimeSeries(name string) int64 {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return 0
	}
	sql := fmt.Sprintf("count timeseries %v", name)
	//fmt.Println("sql:", sql)
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sessionDataSet.Close()
		//fmt.Println()
		sessionDataSet.Next()
		//dataType := sessionDataSet.GetColumnDataType(0)
		columnName := sessionDataSet.GetColumnName(0)
		count := sessionDataSet.GetInt64(columnName)

		return count
	} else {
		log.Println(err)
	}
	return 0
}

func (db *IoTDB) IsDbExist(name string) bool {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return false
	}
	sql := fmt.Sprintf("count databases %v", name)
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sessionDataSet.Close()
		sessionDataSet.Next()
		count := sessionDataSet.GetInt32("count")
		return count == 1
	} else {
		log.Println(err)
	}
	return false
}

func convertTimeSeries(sds *client.SessionDataSet) ([]*monitor.TimeSeries, error) {
	var dataList []*monitor.TimeSeries
	for next, err := sds.Next(); err == nil && next; next, err = sds.Next() {
		d := monitor.TimeSeries{}
		d.Time = sds.GetTimestamp()
		d.TimeSeries = sds.GetText("Timeseries")
		d.Value = sds.GetText("Value")
		d.DataType = sds.GetText("DataType")
		dataList = append(dataList, &d)
	}
	return dataList, nil
}

func convertDataSetToServerData(sds *client.SessionDataSet) ([]*monitor.ServerData, error) {
	var serverData []*monitor.ServerData
	for next, err := sds.Next(); err == nil && next; next, err = sds.Next() {
		d := monitor.ServerData{}
		d.EventTime = sds.GetTimestamp()
		d.Hostname = sds.GetText("hostname")
		d.TotalCPUUsage = sds.GetDouble("total_cpu_usage")
		d.MemUsedPercent = sds.GetDouble("mem_used_percent")
		loggerv.Logger.Info("Read .", zap.Any("ServerData", d))
		serverData = append(serverData, &d)
	}
	loggerv.Logger.Info("Read .", zap.Any("serverData list:", serverData))
	return serverData, nil
}

func printDataSet1(sds *client.SessionDataSet) {
	showTimestamp := !sds.IsIgnoreTimeStamp()
	if showTimestamp {
		fmt.Print("Time\t\t\t\t")
	}

	for i := 0; i < sds.GetColumnCount(); i++ {
		fmt.Printf("--%s\t", sds.GetColumnName(i))
	}
	fmt.Println()

	for next, err := sds.Next(); err == nil && next; next, err = sds.Next() {
		if showTimestamp {
			fmt.Printf("%s\t", sds.GetText(client.TimestampColumnName))
		}
		for i := 0; i < sds.GetColumnCount(); i++ {
			columnName := sds.GetColumnName(i)
			v := sds.GetValue(columnName)
			if v == nil {
				v = "null"
			}
			fmt.Printf("**%v\t\t", v)
		}
		fmt.Println()
	}
}

func (db *IoTDB) ExecuteQueryStatement(sql string) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return
	}
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sessionDataSet.Close()
		printDataSet1(sessionDataSet)
	} else {
		log.Println(err)
	}
}

func (db *IoTDB) CreateMultiTimeSeries() {
	var (
		paths       = []string{"root.sg1.dev1.temperature", "root.sg1.dev1.temperature1"}
		dataTypes   = []client.TSDataType{client.TEXT, client.FLOAT}
		encodings   = []client.TSEncoding{client.PLAIN, client.PLAIN}
		compressors = []client.TSCompressionType{client.SNAPPY, client.SNAPPY}
	)
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		checkError(session.CreateMultiTimeseries(paths, dataTypes, encodings, compressors))
	}

}

func (db *IoTDB) CreateAlignedTimeSeries(prefixPath string, measurements, measurementAlias []string) {
	var (
		dataTypes = []client.TSDataType{
			client.FLOAT,
			client.FLOAT,
		}
		encodings = []client.TSEncoding{
			client.PLAIN,
			client.PLAIN,
		}
		compressors = []client.TSCompressionType{
			client.LZ4,
			client.LZ4,
		}
	)
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		checkError(session.CreateAlignedTimeseries(prefixPath, measurements, dataTypes, encodings, compressors, measurementAlias))
	}

}

func (db *IoTDB) CreateTimeSeries(path string, dataType client.TSDataType, encoding client.TSEncoding, compressor client.TSCompressionType) {
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		checkError(session.CreateTimeseries(path, dataType, encoding, compressor, nil, nil))
	}
}

func (db *IoTDB) DeleteTimeseries(paths ...string) {
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		checkError(session.DeleteTimeseries(paths))
	}

}

func (db *IoTDB) ExecuteQueryStatement1(sql string, resultType interface{}) (interface{}, error) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return nil, err
	}
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err != nil {
		log.Println(err)
		return nil, err
	}
	defer sessionDataSet.Close()
	if err == nil {
		defer sessionDataSet.Close()
		var results []interface{}
		for next, err := sessionDataSet.Next(); err == nil && next; next, err = sessionDataSet.Next() {
			result := reflect.New(reflect.TypeOf(resultType)).Elem()

			for i := 0; i < result.NumField(); i++ {
				field := result.Field(i)
				fieldName := result.Type().Field(i).Tag.Get("json")
				fmt.Println("fieldName --- ", fieldName)

				if fieldName == "" {
					fieldName = result.Type().Field(i).Name
				}

				switch field.Kind() {
				case reflect.String:
					field.SetString(sessionDataSet.GetText(fieldName))
				case reflect.Int64:
					field.SetInt(sessionDataSet.GetTimestamp())
				case reflect.Float64:
					field.SetFloat(sessionDataSet.GetDouble(fieldName))
				case reflect.Uint64:
					value, err := strconv.ParseUint(sessionDataSet.GetText(fieldName), 10, 64)
					if err == nil {
						field.SetUint(value)
					}
					//case reflect.Slice:
					//	if field.Type().Elem().Kind() == reflect.Float64 {
					//		field.Set(reflect.ValueOf(parseCoreUsages(sessionDataSet.GetString(fieldName))))
					//	}
				}
			}
			results = append(results, result.Interface())
		}
		return results, nil
	} else {
		log.Println(err)
	}
	return nil, err
}

func (db *IoTDB) InsertTabletByServerData(path string, data monitor.ServerData) error {
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err == nil {
		if tablet, err := createTabletByServerData(path, data); err == nil {
			status, err := session.InsertTablet(tablet, false)
			tablet.Reset()
			checkError(status, err)
		} else {
			log.Fatal(err)
			return err
		}
	}
	return nil

}

func createTabletByServerData(path string, data monitor.ServerData) (*client.Tablet, error) {
	schemas := []*client.MeasurementSchema{
		{Measurement: "hostname", DataType: client.TEXT},
		{Measurement: "event_time", DataType: client.INT64},
		{Measurement: "total_cpu_usage", DataType: client.DOUBLE},
		{Measurement: "mem_total", DataType: client.INT64},
		{Measurement: "mem_available", DataType: client.INT64},
		{Measurement: "mem_used", DataType: client.INT64},
		{Measurement: "mem_free", DataType: client.INT64},
		{Measurement: "mem_used_percent", DataType: client.DOUBLE},
	}
	for i := range data.CoreUsages {
		schema := &client.MeasurementSchema{Measurement: fmt.Sprintf("core_usages_%d", i), DataType: client.DOUBLE}
		schemas = append(schemas, schema)
		//fmt.Printf("1. i=", i)
		//fmt.Println("")
	}
	loggerv.Logger.Info("get schemas", zap.Any("schemas:", schemas))
	tablet, err := client.NewTablet(path, schemas, 1)
	if err != nil {
		return nil, err
	}
	columnIndex := 0
	tablet.SetTimestamp(data.EventTime, 0)
	tablet.SetValueAt(data.Hostname, columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(data.EventTime, columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(data.TotalCPUUsage, columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(int64(data.MemTotal), columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(int64(data.MemAvailable), columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(int64(data.MemUsed), columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(int64(data.MemFree), columnIndex, 0)
	columnIndex++
	tablet.SetValueAt(data.MemUsedPercent, columnIndex, 0)
	columnIndex++
	for i := range data.CoreUsages {
		//fmt.Printf("2. i=", i)
		//fmt.Println("")
		tablet.SetValueAt(data.CoreUsages[i], columnIndex, 0)
		columnIndex++
	}
	tablet.RowSize++
	return tablet, nil
}

func createTabletByServerDataList(path string, data []monitor.ServerData) (*client.Tablet, error) {
	if data == nil || len(data) == 0 {
		return nil, errors.New("the input array is empty")
	}
	schemas := []*client.MeasurementSchema{
		{Measurement: "hostname", DataType: client.TEXT},
		{Measurement: "event_time", DataType: client.INT64},
		{Measurement: "total_cpu_usage", DataType: client.FLOAT},
		{Measurement: "mem_total", DataType: client.FLOAT},
		{Measurement: "mem_available", DataType: client.FLOAT},
		{Measurement: "mem_used", DataType: client.FLOAT},
		{Measurement: "mem_free", DataType: client.FLOAT},
		{Measurement: "mem_used_percent", DataType: client.FLOAT},
	}
	for i := range data[0].CoreUsages {
		schema := &client.MeasurementSchema{Measurement: fmt.Sprintf("core_usages_%d", i), DataType: client.FLOAT}
		schemas = append(schemas, schema)
	}
	length := len(data)
	tablet, err := client.NewTablet(path, schemas, length)
	if err != nil {
		return nil, err
	}
	columnIndex := 0
	for i := 0; i < length; i++ {
		tablet.SetTimestamp(data[i].EventTime, 0)
		tablet.SetValueAt(data[i].Hostname, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].EventTime, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].TotalCPUUsage, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].MemTotal, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].MemAvailable, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].MemUsed, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].MemFree, columnIndex, 0)
		columnIndex++
		tablet.SetValueAt(data[i].MemUsedPercent, columnIndex, 0)
		columnIndex++
		for _, i := range data[i].CoreUsages {
			tablet.SetValueAt(i, columnIndex, 0)
			columnIndex++
		}
		tablet.RowSize++
	}
	return tablet, nil
}

func createTablet(rowCount int) (*client.Tablet, error) {
	tablet, err := client.NewTablet("root.ln.device1", []*client.MeasurementSchema{
		{
			Measurement: "restart_count",
			DataType:    client.INT32,
		}, {
			Measurement: "price",
			DataType:    client.DOUBLE,
		}, {
			Measurement: "tick_count",
			DataType:    client.INT64,
		}, {
			Measurement: "temperature",
			DataType:    client.FLOAT,
		}, {
			Measurement: "description",
			DataType:    client.TEXT,
		},
		{
			Measurement: "status",
			DataType:    client.BOOLEAN,
		},
	}, rowCount)

	if err != nil {
		return nil, err
	}
	ts := time.Now().UTC().UnixNano() / 1000000
	for row := 0; row < int(rowCount); row++ {
		ts++
		tablet.SetTimestamp(ts, row)
		tablet.SetValueAt(rand.Int31(), 0, row)
		tablet.SetValueAt(rand.Float64(), 1, row)
		tablet.SetValueAt(rand.Int63(), 2, row)
		tablet.SetValueAt(rand.Float32(), 3, row)
		tablet.SetValueAt(fmt.Sprintf("Test Device %d", row+1), 4, row)
		tablet.SetValueAt(bool(ts%2 == 0), 5, row)
		tablet.RowSize++
	}
	return tablet, nil
}

func checkError(status *rpc.TSStatus, err error) {
	if err != nil {
		log.Fatal(err)
	}

	if status != nil {
		if err = client.VerifySuccess(status); err != nil {
			log.Println(err)
		}
	}
}

func (db *IoTDB) QueryServerData(sql string) ([]*monitor.ServerData, error) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return nil, err
	}
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sessionDataSet.Close()
		//printDataSet1(sessionDataSet)
		return convertDataSetToServerData(sessionDataSet)
	} else {
		log.Println(err)
	}
	return nil, err
}
func (db *IoTDB) QueryTimeSeries(sql string) ([]*monitor.TimeSeries, error) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return nil, err
	}
	sds, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sds.Close()
		//printDataSet1(sessionDataSet)
		return convertTimeSeries(sds)
	} else {
		log.Println(err)
	}
	return nil, err
}

func (db *IoTDB) QueryTotalVmRecord(sql string) ([]*model.TotalVmRecord, error) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return nil, err
	}
	sds, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sds.Close()
		//printDataSet1(sessionDataSet)
		return convertTotalVmRecord(sds)
	} else {
		log.Println(err)
	}
	return nil, err
}

func convertTotalVmRecord(sds *client.SessionDataSet) ([]*model.TotalVmRecord, error) {
	var dataList []*model.TotalVmRecord
	for next, err := sds.Next(); err == nil && next; next, err = sds.Next() {
		d := model.TotalVmRecord{}
		d.Time = sds.GetTimestamp()
		d.Device = sds.GetText("Device")
		d.TotalCPUUsage = sds.GetDouble("total_cpu_usage")
		d.MemUsedPercent = sds.GetDouble("mem_used_percent")
		dataList = append(dataList, &d)
	}
	return dataList, nil
}

func (db *IoTDB) QueryLastTs(sql string) ([]*monitor.ServerData, error) {
	var timeout int64 = 1000
	session, err := db.sessionPool.GetSession()
	defer db.sessionPool.PutBack(session)
	if err != nil {
		log.Print(err)
		return nil, err
	}
	sessionDataSet, err := session.ExecuteQueryStatement(sql, &timeout)
	if err == nil {
		defer sessionDataSet.Close()
		//printDataSet1(sessionDataSet)
		return convertDataSetToServerData(sessionDataSet)
	} else {
		log.Println(err)
	}
	return nil, err
}

//func cnvertDataSetToServerData(sds *client.SessionDataSet) (*[]monitor.ServerData, error)
